package com.feeyo.prophet.config;

import jakarta.servlet.http.HttpServletResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


@Slf4j
@Component
public class SseClientConfig {
    private static final Map<String, SseEmitter> SSE_EMITTER_MAP = new ConcurrentHashMap<>();

    @Autowired
    private MongoOperations mongoOperations;
    /**
     * 创建连接
     */
    public SseEmitter getSse(String uid, HttpServletResponse response) {
        // 检查并关闭旧连接
        if (SSE_EMITTER_MAP.containsKey(uid)) {
            log.info("连接已存在，移除旧连接: {}", uid);
            closeSse(uid);
        }
        log.info("开始创建sse连接: {}", uid);
        SseEmitter sseEmitter = new SseEmitter(0L);
        try {
            response.setContentType("text/event-stream");
            response.setCharacterEncoding("UTF-8");

            sseEmitter.onCompletion(() -> {
                log.info("[{}]结束连接", uid);
                cleanupMongoData(uid);
                closeSse(uid); // 自动调用 closeSse
            });

            sseEmitter.onTimeout(() -> {
                log.info("[{}]连接超时", uid);
                closeSse(uid); // 自动调用 closeSse
            });

            sseEmitter.onError(throwable -> {
                log.error("[{}]连接异常: {}", uid, throwable.getMessage());
                closeSse(uid); // 自动调用 closeSse
            });

            log.info("getSse uid:{}", uid);

            sseEmitter.send(SseEmitter.event().reconnectTime(5000));
            SSE_EMITTER_MAP.put(uid, sseEmitter);
            log.info("[{}]创建sse连接成功", uid);
        } catch (IOException e) {
            log.error("SSE连接异常: {}", e.getMessage());
            closeSse(uid); // 确保异常时关闭连接
        } catch (Exception e) {
            log.error("未知异常: {}", e.getMessage());
            closeSse(uid); // 确保异常时关闭连接
        }
        return sseEmitter;
    }

    /**
     * 给指定用户发送消息
     *
     */
    public boolean sendMessage(String uid, String messageId, String message) {
        if (StringUtils.isEmpty(message)) {
            log.info("参数异常，msg为null:{}", uid);
            return false;
        }
        SseEmitter sseEmitter = SSE_EMITTER_MAP.get(uid);
        if (sseEmitter == null) {
            log.info("消息推送失败uid:[{}],没有创建连接，请重试。", uid);
            return false;
        }
        try {
            sseEmitter.send(SseEmitter.event().id(messageId).reconnectTime(1*60*1000L).data(message));
            log.info("用户{},消息id:{},推送成功:{}", uid,messageId, message);
            return true;
        }catch (Exception e) {
            SSE_EMITTER_MAP.remove(uid);
            log.info("用户{},消息id:{},推送异常:{}", uid,messageId, e.getMessage());
            sseEmitter.complete();
            return false;
        }
    }

    /**
     * 判断数据是不是存在
     * @param uid   是不是存在
     */
    public boolean isSseEmitterExist(String uid){
        return SSE_EMITTER_MAP.containsKey(uid);
    }
    private void cleanupMongoData(String uid) {
        // 删除 MongoDB 中与该 uid 相关的数据
        Query query = Query.query(Criteria.where("_id").is(uid));
        mongoOperations.remove(query, "prophet_chat");
        log.info("清理 MongoDB 数据: uid={}", uid);
    }

    /**
     * 断开
     * @param uid
     */
    public void closeSse(String uid) {
        SseEmitter sseEmitter = SSE_EMITTER_MAP.remove(uid);
        if (sseEmitter != null) {
            try {
                sseEmitter.complete();
                log.info("关闭连接: {}", uid);
            } catch (IllegalStateException e) {
                log.warn("连接已关闭或响应已完成: {}", uid);
            }
        }
    }

}